package controller

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"mygame/app/common/ret"
	mongodao "mygame/app/dao/mongoDao"
	mongomodel "mygame/app/model/mongoModel"
	"mygame/app/services"
	"mygame/app/utils"
	"mygame/pkg/logger"
	mysqlpkg "mygame/pkg/mysqlPkg"
	redispkg "mygame/pkg/redisPkg"
	"time"

	"github.com/gin-gonic/gin"
	"github.com/segmentio/kafka-go"
	"github.com/sirupsen/logrus"
	"github.com/streadway/amqp"
)

type Test struct{}

func init() {
	// go tKafkaCunsumer()
	go rabbitMqConsumer()
}

func (t Test) Test(ctx *gin.Context) {
	// testRedis(ctx)
	// mongodbTest(ctx)
	// tlog()

	// tKafka()

	// rabbitMqProducer()

	// tMysql()

	// testService := &services.Test{Name: "哇哈哈123"}
	// testService.Test()

	// httpTest()

	// res, err := mongodao.Counter{}.GetNextID("bb")
	// utils.Println("xxx===>", res, err)

	t.wxTest()

	// hash := md5.Sum([]byte("123456"))
	// md5Str := hex.EncodeToString(hash[:])

	md5Str := utils.Md5("123456")

	ret.Success(ctx, gin.H{
		"get":  ctx.Query("a"),
		"data": md5Str,
	})
	return
}

func (t Test) wxTest() {
	// res, err := services.Weixin{}.GetAccessToken()
	// data := map[string]interface{}{
	// 	"res":     string(res),
	// 	"err":     err,
	// 	"hobbies": []string{"test", "test"},
	// }
	// jsonBytes, _ := json.Marshal(data)
	// fmt.Println("=====0000>>", string(jsonBytes))

	// services.Weixin{}.GetOpenid("0b3pj3ll22IxAf4SsNkl2EwJCZ3pj3ln")

	res, err := services.User{}.OpenidLogin("test010")
	utils.Println("1111xxx===>", res.Token, err)
}

func (t Test) T2(ctx *gin.Context) {
	// var data map[string]interface{}
	// rawData, _ := ctx.GetRawData()
	// json.Unmarshal(rawData, &data)
	ret.Success(ctx, gin.H{
		"get": ctx.Query("kk"),
		// "postRaw":  data,
		"postForm": ctx.PostForm("abc"), //注意 ctx.GetRawData() 后这个拿不到
	})
}

func httpTest() {
	var data map[string]interface{}
	body, err := utils.NewHttpClient("").Post("http://localhost:8081/t2?kk=asdf", map[string]string{"abc": "66667777"}, "form")
	json.Unmarshal(body, &data)
	fmt.Println("===>>", data, err)
}

func tKafka() {
	// to produce messages
	topic := "my-topic"
	partition := 0

	conn, err := kafka.DialLeader(context.Background(), "tcp", "localhost:9092", topic, partition)
	if err != nil {
		log.Fatal("failed to dial leader:", err)
	}

	conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
	_, err = conn.WriteMessages(
		kafka.Message{Value: []byte("one!")},
		kafka.Message{Value: []byte("two!")},
		kafka.Message{Value: []byte("three!")},
	)
	if err != nil {
		log.Fatal("failed to write messages:", err)
	}

	if err := conn.Close(); err != nil {
		log.Fatal("failed to close writer:", err)
	}
}

func tKafkaCunsumer() {
	r := kafka.NewReader(kafka.ReaderConfig{
		Brokers: []string{"localhost:9092"},
		Topic:   "my-topic",
		// GroupID: "your-consumer-group", //设置后就消费不到了
		// CommitInterval: time.Second * 1,
		// StartOffset: kafka.LastOffset,
		MaxBytes: 10e6, // 10MB
	})
	r.SetOffset(kafka.LastOffset) //todo 	缺少offset

	for {
		m, err := r.ReadMessage(context.Background())
		if err != nil {
			break
		}
		fmt.Printf("message at offset %d: %s = %s\n", m.Offset, string(m.Key), string(m.Value))
	}

	if err := r.Close(); err != nil {
		log.Fatal("failed to close reader:", err)
	}
}

func tMysql() {
	mysqlpkg.DB.Exec("INSERT INTO `test` (`name`) VALUES ('test');")
}

func tlog() {
	// {"level":"debug","msg":"map[test3:test4]","test1":"test2","time":"2025-03-19 19:46:28"}
	logger.Debug(logrus.Fields{
		"test1": "test2",
	}, map[string]interface{}{"test3": "test4"})
}

func testRedis(ctx *gin.Context) {
	redispkg.Rdb.Set(redispkg.Ctx, "a", "哇哈哈", 0)
	a := redispkg.Rdb.Get(redispkg.Ctx, "a")
	fmt.Println(a)
	ret.Success(ctx, "abc")
}

func mongodbTest(ctx *gin.Context) {

	// res, err := mgodb.Client.Database("local").Collection("test_user").InsertOne(mgodb.Ctx, map[string]interface{}{"username": "test", "password": "123456"})
	// if err != nil {
	// 	ret.Error(ctx, err.Error())
	// 	return
	// }
	// fmt.Println(res)

	user := mongomodel.User{Username: "test222", Password: "123456"}
	_, err := mongodao.User{}.AddUser(user)
	if err != nil {
		ret.Error(ctx, err.Error())
		return
	}
	ret.Success(ctx, "abc")
}

func rabbitMqConsumer() {
	// 1. 连接 RabbitMQ
	conn, err := amqp.Dial("amqp://admin:admin123@localhost:5672/")
	if err != nil {
		log.Fatalf("连接失败: %v", err)
	}
	defer conn.Close()

	// 2. 创建通道
	ch, err := conn.Channel()
	if err != nil {
		log.Fatalf("通道创建失败: %v", err)
	}
	defer ch.Close()

	// 3. 声明队列（与生产者一致）
	q, err := ch.QueueDeclare(
		"task_queue", // 队列名
		true,         // 持久化
		false,        // 自动删除
		false,        // 排他性
		false,        // 非阻塞
		nil,
	)
	if err != nil {
		log.Fatalf("队列声明失败: %v", err)
	}

	// 4. 设置预取计数（提升并发性能）
	err = ch.Qos(
		1,     // 每个消费者预取消息数
		0,     // 预取字节大小
		false, // 全局模式
	)
	if err != nil {
		log.Fatalf("QoS设置失败: %v", err)
	}

	// 5. 消费消息
	msgs, err := ch.Consume(
		q.Name, // 队列名
		"",     // 消费者标识
		false,  // 手动确认（关键防消息丢失）
		false,  // 排他性
		false,  // 非阻塞
		false,  // 额外属性
		nil,
	)
	if err != nil {
		log.Fatalf("消费者注册失败: %v", err)
	}

	for d := range msgs {
		log.Printf("收到消息: %s", d.Body)
		time.Sleep(time.Second) // 模拟耗时操作
		d.Ack(false)            // 手动确认
	}
}

func rabbitMqProducer() {
	// 1. 连接 RabbitMQ
	conn, err := amqp.Dial("amqp://admin:admin123@localhost:5672/")
	if err != nil {
		log.Fatalf("连接失败: %v", err)
	}
	defer conn.Close()

	// 2. 创建通道
	ch, err := conn.Channel()
	if err != nil {
		log.Fatalf("通道创建失败: %v", err)
	}
	defer ch.Close()

	// 3. 声明队列（持久化）
	q, err := ch.QueueDeclare(
		"task_queue", // 队列名
		true,         // 持久化
		false,        // 自动删除
		false,        // 排他性
		false,        // 非阻塞
		nil,          // 额外参数
	)
	if err != nil {
		log.Fatalf("队列声明失败: %v", err)
	}

	// 4. 发送消息
	body := "Hello RabbitMQ!"
	err = ch.Publish(
		"",     // 交换机（默认）
		q.Name, // 路由键
		false,  // 强制路由
		false,  // 立即发送
		amqp.Publishing{
			DeliveryMode: amqp.Persistent, // 消息持久化
			ContentType:  "text/plain",
			Body:         []byte(body),
		},
	)
	if err != nil {
		log.Fatalf("消息发送失败: %v", err)
	}
	log.Printf(" [x] 已发送: %s", body)
}
